package FutureAndPromise;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;

/**
 * Demo02 - JDK CompletableFuture 详解（Java 8+）
 * 
 * 学习目标：
 * 1. 理解 CompletableFuture 的优势
 * 2. 掌握回调方法的使用
 * 3. 学习异步操作的组合
 * 4. 了解异常处理机制
 */
@Slf4j
public class Demo02_CompletableFuture {

    public static void main(String[] args) throws Exception {
        log.info("=== CompletableFuture 详解 ===\n");
        
        demo1_BasicUsage();
        demo2_Callbacks();
        demo3_Chaining();
        demo4_Combining();
        demo5_ExceptionHandling();
        demo6_AllOfAnyOf();
        
        // 等待所有异步任务完成
        Thread.sleep(3000);
    }

    /**
     * Demo 1: CompletableFuture 基础用法
     */
    private static void demo1_BasicUsage() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 1: CompletableFuture 基础用法");
        log.info("└─────────────────────────────────────────\n");
        
        // 1. 创建已完成的 Future
        CompletableFuture<String> completedFuture = 
            CompletableFuture.completedFuture("已完成的结果");
        log.info("1. 已完成的 Future: {}", completedFuture.get());
        
        // 2. 异步执行任务（使用默认线程池 ForkJoinPool）
        CompletableFuture<String> asyncFuture = CompletableFuture.supplyAsync(() -> {
            log.info("   在线程 {} 中执行", Thread.currentThread().getName());
            try {
                Thread.sleep( 500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "异步执行的结果";
        });
        
        log.info("2. 异步任务已提交，立即返回");
        String result = asyncFuture.get(); // 阻塞等待结果
        log.info("   异步任务完成: {}", result);
        
        // 3. 使用自定义线程池
        ExecutorService executor = Executors.newFixedThreadPool(2);
        CompletableFuture<String> customFuture = CompletableFuture.supplyAsync(() -> {
            log.info("   在自定义线程池中执行: {}", Thread.currentThread().getName());
            return "自定义线程池的结果";
        }, executor);
        
        log.info("3. 使用自定义线程池: {}", customFuture.get());
        
        executor.shutdown();
    }

    /**
     * Demo 2: 回调方法（Callbacks）
     */
    private static void demo2_Callbacks() {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 2: 回调方法");
        log.info("└─────────────────────────────────────────\n");
        
        // 1. thenApply - 转换结果（有返回值）
        log.info("1. thenApply - 转换结果");
        CompletableFuture.supplyAsync(() -> {
            log.info("   步骤1: 获取数据...");
            return "Hello";
        }).thenApply(s -> {
            log.info("   步骤2: 转换数据 '{}' -> '{} World'", s, s);
            return s + " World";
        }).thenApply(s -> {
            log.info("   步骤3: 转换为大写 '{}' -> '{}'", s, s.toUpperCase());
            return s.toUpperCase();
        }).thenAccept(result -> {
            log.info("   最终结果: {}", result);
        });
        
        // 2. thenAccept - 消费结果（无返回值）
        log.info("\n2. thenAccept - 消费结果");
        CompletableFuture.supplyAsync(() -> "数据")
            .thenAccept(data -> {
                log.info("   接收到数据: {}", data);
                log.info("   处理数据...");
            });
        
        // 3. thenRun - 执行操作（不接收参数，无返回值）
        log.info("\n3. thenRun - 执行后续操作");
        CompletableFuture.supplyAsync(() -> {
            log.info("   异步任务执行中...");
            return "完成";
        }).thenRun(() -> {
            log.info("   任务完成后的清理工作");
        });
        
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * Demo 3: 链式调用
     */
    private static void demo3_Chaining() {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 3: 链式调用");
        log.info("└─────────────────────────────────────────\n");
        
        // 模拟一个完整的业务流程
        log.info("模拟：用户注册流程");
        
        CompletableFuture.supplyAsync(() -> {
            // 步骤1: 验证用户输入
            log.info("   [1/4] 验证用户输入...");
            sleep(200);
            return "user123";
        }).thenApply(username -> {
            // 步骤2: 检查用户名是否存在
            log.info("   [2/4] 检查用户名 '{}' 是否可用...", username);
            sleep(300);
            return username + "@example.com";
        }).thenApply(email -> {
            // 步骤3: 创建用户账户
            log.info("   [3/4] 创建账户，邮箱: {}", email);
            sleep(400);
            return new User("user123", email);
        }).thenAccept(user -> {
            // 步骤4: 发送欢迎邮件
            log.info("   [4/4] 发送欢迎邮件到: {}", user.email);
            log.info("   ✅ 用户注册完成: {}", user);
        });
        
        try {
            Thread.sleep(1500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * Demo 4: 组合多个 Future
     */
    private static void demo4_Combining() {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 4: 组合多个 Future");
        log.info("└─────────────────────────────────────────\n");
        
        // 1. thenCompose - 串行组合（依赖前一个结果）
        log.info("1. thenCompose - 串行组合");
        CompletableFuture.supplyAsync(() -> {
            log.info("   获取用户ID...");
            sleep(200);
            return "user123";
        }).thenCompose(userId -> {
            // 使用 userId 获取用户详情
            log.info("   根据ID '{}' 获取用户详情...", userId);
            return CompletableFuture.supplyAsync(() -> {
                sleep(300);
                return new User(userId, userId + "@example.com");
            });
        }).thenAccept(user -> {
            log.info("   ✅ 最终用户信息: {}", user);
        });
        
        sleep(600);
        
        // 2. thenCombine - 并行组合（两个 Future 都完成后合并）
        log.info("\n2. thenCombine - 并行组合");
        
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            log.info("   任务1: 查询用户信息...");
            sleep(300);
            return "User{name='张三'}";
        });
        
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            log.info("   任务2: 查询订单信息...");
            sleep(400);
            return "Order{count=5}";
        });
        
        future1.thenCombine(future2, (user, order) -> {
            log.info("   ✅ 合并结果: {} - {}", user, order);
            return user + " - " + order;
        });
        
        sleep(600);
        
        // 3. thenAcceptBoth - 并行组合（消费两个结果）
        log.info("\n3. thenAcceptBoth - 并行消费");
        
        CompletableFuture<Integer> priceTask = CompletableFuture.supplyAsync(() -> {
            log.info("   计算价格...");
            sleep(200);
            return 100;
        });
        
        CompletableFuture<Integer> discountTask = CompletableFuture.supplyAsync(() -> {
            log.info("   计算折扣...");
            sleep(250);
            return 20;
        });
        
        priceTask.thenAcceptBoth(discountTask, (price, discount) -> {
            int finalPrice = price - discount;
            log.info("   ✅ 原价: {}, 折扣: {}, 最终价格: {}", price, discount, finalPrice);
        });
        
        sleep(500);
    }

    /**
     * Demo 5: 异常处理
     */
    private static void demo5_ExceptionHandling() {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 5: 异常处理");
        log.info("└─────────────────────────────────────────\n");
        
        // 1. exceptionally - 捕获异常并提供默认值
        log.info("1. exceptionally - 捕获异常");
        CompletableFuture.supplyAsync(() -> {
            log.info("   执行可能失败的任务...");
            if (Math.random() > 0.5) {
                throw new RuntimeException("任务执行失败！");
            }
            return "成功结果";
        }).exceptionally(ex -> {
            log.error("   ❌ 捕获异常: {}", ex.getMessage());
            return "默认值";
        }).thenAccept(result -> {
            log.info("   最终结果: {}", result);
        });
        
        sleep(300);
        
        // 2. handle - 同时处理成功和失败
        log.info("\n2. handle - 处理成功和失败");
        CompletableFuture.supplyAsync(() -> {
            log.info("   执行任务...");
            if (System.currentTimeMillis() % 2 == 0) {
                throw new RuntimeException("模拟异常");
            }
            return "成功";
        }).handle((result, ex) -> {
            if (ex != null) {
                log.error("   ❌ 处理失败: {}", ex.getMessage());
                return "失败后的恢复值";
            } else {
                log.info("   ✅ 处理成功: {}", result);
                return result;
            }
        }).thenAccept(finalResult -> {
            log.info("   最终处理结果: {}", finalResult);
        });
        
        sleep(300);
        
        // 3. whenComplete - 完成后的回调（不影响结果）
        log.info("\n3. whenComplete - 完成后的回调");
        CompletableFuture.supplyAsync(() -> {
            log.info("   执行任务...");
            return "任务结果";
        }).whenComplete((result, ex) -> {
            if (ex != null) {
                log.error("   任务失败: {}", ex.getMessage());
            } else {
                log.info("   任务成功: {}", result);
            }
            log.info("   执行清理工作...");
        }).thenAccept(result -> {
            log.info("   继续处理: {}", result);
        });
        
        sleep(300);
    }

    /**
     * Demo 6: allOf 和 anyOf
     */
    private static void demo6_AllOfAnyOf() {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 6: allOf 和 anyOf");
        log.info("└─────────────────────────────────────────\n");
        
        // 1. allOf - 等待所有 Future 完成
        log.info("1. allOf - 等待所有任务完成");
        
        CompletableFuture<String> task1 = CompletableFuture.supplyAsync(() -> {
            log.info("   任务1 开始...");
            sleep(300);
            log.info("   任务1 完成");
            return "结果1";
        });
        
        CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> {
            log.info("   任务2 开始...");
            sleep(500);
            log.info("   任务2 完成");
            return "结果2";
        });
        
        CompletableFuture<String> task3 = CompletableFuture.supplyAsync(() -> {
            log.info("   任务3 开始...");
            sleep(200);
            log.info("   任务3 完成");
            return "结果3";
        });
        
        CompletableFuture<Void> allTasks = CompletableFuture.allOf(task1, task2, task3);
        
        allTasks.thenRun(() -> {
            log.info("   ✅ 所有任务都完成了！");
            try {
                log.info("   收集结果: {}, {}, {}", 
                    task1.get(), task2.get(), task3.get());
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        
        sleep(800);
        
        // 2. anyOf - 任意一个 Future 完成即可
        log.info("\n2. anyOf - 任意任务完成");
        
        CompletableFuture<String> fast = CompletableFuture.supplyAsync(() -> {
            log.info("   快速任务 开始...");
            sleep(100);
            log.info("   快速任务 完成");
            return "快速结果";
        });
        
        CompletableFuture<String> slow = CompletableFuture.supplyAsync(() -> {
            log.info("   慢速任务 开始...");
            sleep(500);
            log.info("   慢速任务 完成");
            return "慢速结果";
        });
        
        CompletableFuture<Object> anyTask = CompletableFuture.anyOf(fast, slow);
        
        anyTask.thenAccept(result -> {
            log.info("   ✅ 第一个完成的任务结果: {}", result);
        });
        
        sleep(600);
    }

    // 辅助方法
    private static void sleep(int millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    // 用户类
    static class User {
        String username;
        String email;
        
        User(String username, String email) {
            this.username = username;
            this.email = email;
        }
        
        @Override
        public String toString() {
            return "User{username='" + username + "', email='" + email + "'}";
        }
    }
}

/*
=== 运行结果分析 ===

1. CompletableFuture 的优势：

   ✅ 支持回调（不阻塞）
   ✅ 支持链式调用
   ✅ 支持组合操作
   ✅ 强大的异常处理
   ✅ 可以使用自定义线程池

2. 核心方法分类：

   创建 Future：
   - completedFuture(value)
   - supplyAsync(supplier)
   - supplyAsync(supplier, executor)
   - runAsync(runnable)

   转换结果：
   - thenApply(function)      - 有输入，有输出
   - thenAccept(consumer)     - 有输入，无输出
   - thenRun(runnable)        - 无输入，无输出

   组合 Future：
   - thenCompose(function)    - 串行
   - thenCombine(other, fn)   - 并行合并
   - allOf(futures...)        - 等待全部
   - anyOf(futures...)        - 等待任一

   异常处理：
   - exceptionally(function)  - 捕获异常
   - handle(biFunction)       - 处理结果和异常
   - whenComplete(biConsumer) - 完成回调

3. 异步 vs 同步方法：

   同步方法：在当前线程执行
   - thenApply
   - thenAccept
   - thenRun

   异步方法：在线程池中执行
   - thenApplyAsync
   - thenAcceptAsync
   - thenRunAsync

4. 线程池选择：

   默认：ForkJoinPool.commonPool()
   自定义：传入 Executor 参数

=== 关键知识点 ===

1. CompletableFuture 是 JDK 8 引入的现代异步编程工具
2. 支持函数式编程风格
3. 可以避免回调地狱（Callback Hell）
4. 异常会在链条中传播
5. 使用 allOf/anyOf 处理多个 Future

=== 最佳实践 ===

✅ 使用 thenXxxAsync 避免阻塞调用线程
✅ 使用自定义线程池避免共享 ForkJoinPool
✅ 使用 handle 或 whenComplete 确保资源清理
✅ 避免在 CompletableFuture 链中使用 get()
✅ 合理使用 allOf/anyOf 处理批量任务

❌ 不要在回调中执行阻塞操作
❌ 不要忽略异常处理
❌ 不要过度嵌套（保持链条平坦）
❌ 注意线程池的生命周期管理
*/

